package com.yanduo.report

import com.yanduo.report.AppAnalyseRpt.args
import com.yanduo.utils.JedisPools
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * 媒体报表改造-字典存入到redis中
  *https://www.bilibili.com/video/BV1F4411i7jK?p=24
  *
  * @author Gerry chan
  * 2020/5/4 14:39
  * @version 1.0
  */
object AppDict2Redis {
  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      println(
        """
          |com.yanduo.report.AppDict2Redis
          |参数：
          |  输入路径
        """.stripMargin)
      sys.exit()
    }
    val Array(appdictInputPath) = args

    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")

    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val spark = new SparkSession(sc)

    sc.textFile(appdictInputPath).map(line => {
      val fields = line.split("\t", -1)
      (fields(4), fields(1))
    }).foreachPartition(iter => {
      // 数据存储到redis
      val jedis = JedisPools.getJedis
      iter.foreach(t => {
        jedis.set(t._1, t._2)
      })
      jedis.close()
    })

    sc.stop()
    spark.stop()
  }
}
